package com.blog.wu.myblog.rocketmq;


import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @Author wuyanshen
 * @Date 2018-12-26 2:21 PM
 * @Discription RockerMQ消费者
 */
public class RocketMQConsumer {

    public static void main(String[] args) throws Exception {
        //消费者的组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rmq-group");
        consumer.setInstanceName("rmq-instance");
        //指定NameServer地址，多个地址以 ; 隔开
        consumer.setNamesrvAddr("192.168.0.121:9876");
        //订阅PushTopic下Tag为push的消息
        consumer.subscribe("myblog-rmq","push");

        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt message : list) {
                    String msgId = message.getMsgId();
                    message.getBody();
                    System.out.println("msgId:"+msgId+" msgBody="+message.getBody().toString());
                }
                //消费状态 1.消费成功2.消费失败
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }
}
